using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Query;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Model.Thrift
{
    /// <summary>
    /// AbstractThriftCountQuery is the base query class for all count queries. It uses
    /// lambda expressions instead of anonymous inner classes java Hector. In some ways,
    /// using Lambda is neater than using anonymous inner classes.
    /// </summary>
    /// <typeparam name="K"></typeparam>
    /// <typeparam name="N"></typeparam>
    public abstract class AbstractThriftCountQuery<K, N> : Query<int>
	{
        public AbstractThriftCountQuery(Keyspace k, Serializer<K> keySerializer,
            Serializer<N> nameSerializer)
        {
            if (k == null || nameSerializer == null || keySerializer == null)
            {
                throw new NullReferenceException("Keyspace, Key serializer and Name serializer cannot be null");
            }
            Keyspace = (ExecutingKeyspace)k;
            KeySerializer = keySerializer;
            SlicePredicate = new HSlicePredicate<N>(nameSerializer);
        }

        protected ExecutingKeyspace Keyspace
        {
            get;
            set;
        }
        public string ColumnFamily
        {
            get;set;
        }
        public K Key
        {
            get;set;
        }
        protected Serializer<K> KeySerializer
        {
            get;set;
        }
        protected HSlicePredicate<N> SlicePredicate
        {
            get;
            set;
        }
        public Query<int> SetKey(K key)
        {
            Key = key;
            return this;
        }
        public Query<int> SetColumnFamily(string cf)
        {
            ColumnFamily = cf;
            return this;
        }
        protected  QueryResult<int> CountColumns()
        {
            if (Key == null || ColumnFamily == null)
            {
                throw new NullReferenceException("Key and ColumnFamily cannot be null");
            }
            Func<KeyspaceService, int> x = (KeyspaceService ks) =>
            {
                int count = ks.GetCount(KeySerializer.ToByteBuffer(Key).GetBuffer(),
                    new ColumnParent(ColumnFamily),SlicePredicate.ToThrift());
                return count;
            };
            return new QueryResultImpl<int>(
                Keyspace.DoExecute(new KeyspaceOperationCallbackBuilder<int>(x)), this);
        }
        public Query<int> SetColumnNames(List<N> columnNames)
        {
            SlicePredicate.SetColumnNames(columnNames);
            return this;
        }
        public Query<int> SetRange(N start, N finish, int count)
        {
            SlicePredicate.SetRange(start, finish, false, count);
            return this;
        }
        public abstract QueryResult<int> Execute();
	}
}